﻿// Copyright (c) Microsoft Open Technologies, Inc. All rights reserved. 
// See License.txt in the project root for license information.
using System;
using System.Collections.Generic;
using System.Diagnostics.Contracts;
using System.Fabric;
using System.Linq;
using System.Text;
using System.Threading.Tasks;

namespace System.Threading.Actors.Utilities
{
    // Encapsulates logic for creation and deletion of other actors from an IActorState implementation,
    // as well as acquiring a proxy to another actor.
    public class FabricActorInteractionHelper : IActorInteractor
    {
        // Caches proxies to other actors so that they do not need to be re-created every time they are used.
        // Key is the actor name.  Values are either a FabricActorProxy (non-partitioned service), or a 
        // Dictionary<String, FabricActorProxy> for mapping partition names to proxies.
        private readonly Dictionary<string, object> m_clientsToOtherActors = new Dictionary<string, object>();

        // Cache this
        private readonly static char[] s_creationStringSeparators = new char[] { ';' };

        // actorName needs to be full actor name, e.g. "fabric:/actor/adhoc/myNewActor"
        // Sample creation string would be "applicationType=fabric:/actor/adhoc;serviceType=EmptyActorReplica;numReplicas=3"
        public bool CreateActor(string actorName, string creationString, int numPartitions)
        {
            Contract.Requires(!String.IsNullOrWhiteSpace(actorName));
            Contract.Requires(!String.IsNullOrWhiteSpace(creationString));
            Contract.Requires(Contract.ForAll(FabricActorClient.ReservedDelimiters, reservedChar => !actorName.Contains(reservedChar)), "Actor names cannot contain reserved delimiters like '$'.");
            Contract.Requires(numPartitions >= 1);

            int numReplicas = 2;
            string actorApplicationType = null;
            string actorServiceType = null;
            bool actorAlreadyExisted = false;

            string[] parts = creationString.Split(s_creationStringSeparators, StringSplitOptions.RemoveEmptyEntries);

            foreach (var part in parts)
            {
                string[] keyAndValue = part.Split('=');
                if (keyAndValue.Length != 2)
                    throw new InvalidOperationException("Key-value pairs in creation string must be separated by ':'");

                var key = keyAndValue[0];
                var value = keyAndValue[1];

                if (key.Equals("applicationType"))
                {
                    actorApplicationType = value;
                }
                else if (key.Equals("serviceType"))
                {
                    actorServiceType = value;
                }
                else if (key.Equals("numReplicas"))
                {
                    if (!Int32.TryParse(value, out numReplicas))
                    {
                        throw new InvalidOperationException("Could not parse numReplicas value as an integer");
                    }
                    if (numReplicas < 1)
                        throw new InvalidOperationException(String.Format("Number of replicas must be one or more.  Number specified: {0}", numReplicas));
                }
                else throw new InvalidOperationException("Unknown key " + key);
            }

            if (actorApplicationType == null)
                throw new ArgumentException("creationString must contain a definition for applicationType", "creationString");

            if (actorServiceType == null)
                throw new ArgumentException("creationString must contain a definition for serviceType", "creationString");

            // TODO: Put in a try/catch to handle pre-existing actor gracefully
            try
            {
                FabricActorClient.CreateActor(null, actorApplicationType, actorName, actorServiceType, numReplicas, numPartitions);
            }
            catch (FabricElementAlreadyExistsException)
            {
                // Just swallow these
                actorAlreadyExisted = true;
            }

            return actorAlreadyExisted;
        }

        public void DeleteActor(string actorName)
        {
            // First, dispose of any cached proxies to this actor.
            // I'm assuming that this is running under state lock, and thus needs no special locking.
            FabricActorProxy proxy;

            // We don't know whether this actor is partitioned or not.  For performance, assume it isn't 
            // and do the obvious thing.  If it is partitioned, go down a slower path.
            var key = Tuple.Create<String, String>(actorName, null);
            Object temp = null;
            if (m_clientsToOtherActors.TryGetValue(actorName, out temp))
            {
                proxy = temp as FabricActorProxy;
                if (proxy != null)
                {
                    proxy.Dispose();
                }
                else {
                    var partitionedDictionary = (Dictionary<String, FabricActorProxy>)temp;
                    foreach(var partitionedClient in partitionedDictionary.Values)
                        partitionedClient.Dispose();
                    partitionedDictionary.Clear();
                }
                m_clientsToOtherActors.Remove(actorName);
            }

            // Then delete the actor
            FabricActorClient.DeleteActor(null, actorName);
        }

        public IActorProxy GetActorProxy(string actorName, IActorState sourceState, string targetPartitionName, string myPartitionName)
        {
            return GetActorProxyAsync(actorName, sourceState, targetPartitionName, myPartitionName).Result;
        }

        public async Task<IActorProxy> GetActorProxyAsync(string actorName, IActorState sourceState, string targetPartitionName, string myPartitionName)
        {
            FabricActorProxy result = null;
            Dictionary<String, FabricActorProxy> partitionedDictionary = null;
            Object temp = null;

            // Look up a proxy to this actor.
            if (m_clientsToOtherActors.TryGetValue(actorName, out temp))
            {
                if (targetPartitionName == null)
                    result = (FabricActorProxy)temp;
                else
                {
                    partitionedDictionary = (Dictionary<String, FabricActorProxy>)temp;
                    partitionedDictionary.TryGetValue(targetPartitionName, out result);
                }
            }

            if (result == null)
            {           
                // We are under lock, and constructing a FabricActorProxy will result in several awaits.  And it is not
                // wise to await under lock.  So we want to do this construction in a separate thread in order to 
                // escape the lock.
                //result = Task.Factory.StartNew(() =>
                //    {
                //        return new FabricActorProxy(actorName, sourceState, targetPartitionName, myPartitionName);
                //    }, TaskCreationOptions.PreferFairness).Result;

                // We are no longer under lock...
                result = await FabricActorProxy.CreateAsync(actorName, sourceState, targetPartitionName, myPartitionName);
                //result = new FabricActorProxy(actorName, sourceState, targetPartitionName, myPartitionName);

                // Store the proxy to this actor.
                if (targetPartitionName == null)
                    m_clientsToOtherActors.Add(actorName, result);
                else
                {
                    if (partitionedDictionary == null)
                    {
                        partitionedDictionary = new Dictionary<string, FabricActorProxy>();
                        m_clientsToOtherActors.Add(actorName, partitionedDictionary);
                    }
                    partitionedDictionary.Add(targetPartitionName, result);
                }
            }

            return result;
        }

        public void Dispose()
        {
            // Dispose any remaining clients to other actors
            foreach (Object value in m_clientsToOtherActors.Values)
            {
                FabricActorProxy proxy = value as FabricActorProxy;
                if (proxy != null)
                {
                    try { proxy.Dispose(); } // Do not let any exceptions derail us
                    catch { }
                }
                else
                {
                    var partitionedDictionary = (Dictionary<String, FabricActorProxy>)value;
                    foreach(var partitionedProxy in partitionedDictionary.Values)
                    {
                        try { partitionedProxy.Dispose(); } // Do not let any exceptions derail us
                        catch { }
                    }
                }
            }

            m_clientsToOtherActors.Clear();
        }
    }
}
